package org.shuzhou.g_window;

import java.io.IOException;
import java.time.LocalDate;
import java.time.Period;
import java.time.format.DateTimeFormatter;

import org.shuzhou.model.PersonalInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * 用到的数据集为data/people.txt
 * 陈巧荷,男,汉族,210100195001010014,19500101,132876025675,6spjb0i8d7@qq.com,贵州省黔南布依族苗族自治州平塘县HAU路945号
 * 聂菊天,女,纳西族,232331195001010015,19500101,130346448424,gcfgp6zexi@163.com,广东省韶关市市辖区IHH路154号
 * 祝永峰,女,汉族,142629195001010018,19500101,151088243213,5r26df666n@sina.com,甘肃省酒泉地区MPF路792号
 * 
 * 功能：使用窗口，对各个民族的人员年龄进行累加
 */
public class WinPersonStateful {
    public static void main(String[] args) throws Exception {

        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 获取数据，数据源为Kafka，避免从端口获取中文数据乱码
         * 数据格式
         * 王矗馨,女,汉族,411424195001010028,19500101,150977754190,kk4jlw850x@sina.com,甘肃省临夏回族自治州和政县CNR路884号
         */
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("192.168.56.151:9092,192.168.56.152:9092,192.168.56.153:9092")
                .setTopics("persion")
                .setGroupId("root")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> people = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        /**
         * 获取数据，数据源为kafka
         * 数据格式
         * 王矗馨,女,汉族,411424195001010028,19500101,150977754190,kk4jlw850x@sina.com,甘肃省临夏回族自治州和政县CNR路884号
         */
        SingleOutputStreamOperator<PersonalInfo> ds = people.map(new MapFunction<String, PersonalInfo>() {
            @Override
            public PersonalInfo map(String value) throws Exception {
                // System.out.println(value.toString());
                String encode = new String(value.getBytes("UTF-8"), "UTF-8");
                String[] fields = encode.split(",");
                PersonalInfo info = new PersonalInfo(fields[0], fields[1], fields[2], fields[3], fields[4], fields[5],
                        fields[6], fields[7]);
                return info;
            }
        });

        /**
         * 先使用map计算每个人的年龄，得到元组(民族,age)
         * 然后使用keyBy对民族进行分组
         */
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> keyedStream = ds
                .map(new MapFunction<PersonalInfo, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(PersonalInfo personalInfo) throws Exception {
                        // 通过生日计算年龄
                        long age = calculateAge(personalInfo.getDateOfBirth());
                        return Tuple2.of(personalInfo.getEthnicity(), (int) age);
                    }
                })
                .keyBy(key -> key.f0.toString())
                // .countWindow(4,3)
                // 添加窗口，使用processing-time时间
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                ;


        /**
         * reduce聚合案例：对年龄进行累加
         * 使用有状态算子 keyed state
         */
        SingleOutputStreamOperator<Tuple2<String, Integer>> res = keyedStream.reduce(new RichReduceFunction<Tuple2<String, Integer>>() {
            // sum用于记录状态：年龄总和
            private transient ValueState<Integer> sum;

            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                
                // 获取sum状态
                Integer currentSum = sum.value();

                // 初始化currentSum的值
                if (currentSum == null) {
                    currentSum = 0;
                    currentSum += value1.f1;
                }

                // 每次将输入的value2的年龄进行求和
                if (!value2.equals(null)) {
                    currentSum += value2.f1;
                }

                // 更新sum值
                sum.update(currentSum);

                return Tuple2.of(value1.f0, currentSum);
            }

            /**
             * 使用ValueStateDescriptor构造ValueState<Integer> sum
             * @param config
             * @throws IOException
             */
            @Override
            public void open(Configuration config) throws IOException{
                // 构造时不初始化默认值
                ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>(
                        "sum", // the state name
                        TypeInformation.of(new TypeHint<Integer>() {
                        })); 
                sum = getRuntimeContext().getState(descriptor);
                // 在下面的代码中，判断如果state没有初始化则赋为默认值
                if (sum == null) {
                    sum.update(0);
                }
            }

        });

        res.print();

        env.execute("Flink Operator Example");
    }

    // 计算年龄的函数
    public static int calculateAge(String dateOfBirth) {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
        LocalDate birthDate = LocalDate.parse(dateOfBirth, formatter);
        LocalDate currentDate = LocalDate.now();
        Period period = Period.between(birthDate, currentDate);
        return period.getYears();
    }
}
